package com.zhxin.nettylab.reactor.chapter2;

import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.ReplayingDecoder;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName MultiThreadEchoServerReactor
 * @Description // 多线程Reactor反应器 demo 服务端
 * @Author singleZhang
 * @Email 405780096@qq.com
 * @Date 2020/12/4 0004 下午 4:25
 **/
public class MultiThreadEchoServerReactor {
    private ServerSocketChannel serverSocket;
    private AtomicInteger next = new AtomicInteger(0);
    //selectors集合,引入多个selector选择器
    Selector[] selectors = new Selector[2];
    //引入多个子反应器
    private SubReactor[] subReactors = null;

    private MultiThreadEchoServerReactor() throws IOException {

        //初始化多个选择器
        selectors[0] = Selector.open();
        selectors[1] = Selector.open();

        serverSocket = ServerSocketChannel.open();
        InetSocketAddress address = new InetSocketAddress(8989);
        serverSocket.socket().bind(address);
        serverSocket.configureBlocking(false);//非阻塞

        //让第一个选择器负责监听 新连接事件
        SelectionKey sk = serverSocket.register(selectors[0],SelectionKey.OP_ACCEPT);

        //为选择键绑定handler
        sk.attach(new AcceptHandler());

        //一个子选择器负责一个反应器
        SubReactor subReactor1 = new SubReactor(selectors[0]);
        SubReactor subReactor2 = new SubReactor(selectors[1]);
        subReactors = new SubReactor[]{subReactor1,subReactor2};
    }

    private void startService(){
        //一个子选择器 对应一个线程
        new Thread(subReactors[0]);
        new Thread(subReactors[1]);
    }

    /**
     * 新连接 处理器
     * */
    class AcceptHandler implements Runnable{

        public void run(){
            try{
                //接受新连接,进行业务处理
                SocketChannel socket = serverSocket.accept();
                if(socket != null){
                    // MultiThreadEchoHandler
                    new MultiThreadEchoHandler(selectors[next.get()],socket);
                }
                if(next.incrementAndGet() == selectors.length){
                    next.set(0);
                }
            }catch (Exception e){
                /*...*/
            }
        }
    }

    /**
     * 子反应器
     * */
    static class SubReactor implements Runnable{
        //每子选择器对应的线程负责一个选择器的查询和选择
        final Selector selector;

        SubReactor(Selector selector){
            this.selector = selector;
        }

        public void run(){
           try{
               while (!Thread.interrupted()){
                   selector.select();
                   Set<SelectionKey> selectionKeys = selector.selectedKeys();
                   Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
                   while(keyIterator.hasNext()){
                       SelectionKey key = keyIterator.next();
                       dispatch(key);
                   }

                   selectionKeys.clear();
               }
           }catch (Exception e){
               /* ... */
           }
        }

        /**
         * 反应器的分发
         * */
        private void dispatch(SelectionKey key){
            //调用之前绑定的Handler
            Runnable rKey = (Runnable) key.attachment();
            if(rKey != null){
                rKey.run();
            }
        }
    }

    public static void main(String[] args) throws IOException {
        MultiThreadEchoServerReactor server = new MultiThreadEchoServerReactor();
        server.startService();
    }
}
